首页    技术博客    PHP教程    数据库技术    前端开发   HTML5    Nginx    php论坛
新用户注册 | 会员登录
加入会员，解锁专属内容
取消
热门标签 | HotTags
当前位置:  开发笔记 > 编程语言 > 正文

租户|发生_02RabbitMQ基础

作者：冠臻祥勇8522 | 来源：互联网 | 2023-09-18 20:56
篇首语：本文由编程笔记#小编为大家整理，主要介绍了02RabbitMQ基础相关的知识，希望对你有一定的参考价值。目录0

篇首语：本文由编程笔记#小编为大家整理，主要介绍了02 RabbitMQ基础相关的知识，希望对你有一定的参考价值。


目录

02 RabbitMQ基础

Pt1 工作模型

Pt1.1 Broker

Pt1.2 Exchange

(1) Direct 直连模式

(2) Topic 主题模式

(3) Fanout广播模式

Pt1.3 Queue

Pt1.4 Message

Pt1.5 Connection

Pt1.6 Channel

Pt1.7 VHost

Pt1.8 Consumer

Pt1.9 Producer

Pt2 安装操作

Pt2.1 Windows安装RabbitMQ

(1) Erlang环境说明

(2) 下载安装Erlang

(3) 下载安装RabbitMQ

(4) 启用RabbitMQ插件

(5) 启动RabbitMQ

(6) RabbitMQ管理控台

(7) 初始化RabbitMQ

Pt2.2 CentOS7安装RabbitMQ集群

(1) 安装Docker环境

(2) 安装RabbitMQ

Pt3 基本使用

Pt3.1 基于JAVAAPI示例

Pt3.2 Direct模式

Pt3.3 Topic模式

Pt3.4 Fanout模式



02 RabbitMQ基础

RabbitMQ是一套开源&＃xff08;MPL&＃xff09;的消息队列服务软件&＃xff0c;他是基于AMQP协议的开源实现&＃xff0c;由以高性能、健壮以及可伸缩性出名的 Erlang 语言编写而成。除了AMQP协议外&＃xff0c;RabbitMQ还支持STOMP、MQTT、HTTP和WebSockets等协议。



Pt1 工作模型

由于RabbitMQ是基于AMQP的实现&＃xff0c;服务端的工作模型也是基于AMQP的工作模型&＃xff0c;先来看看RabbitMQ的工作模型&＃xff08;如下图&＃xff09;&＃xff0c;再来介绍RabbitMQ各个模块。

在整个工作模型中&＃xff0c;包含以下几个组件&＃xff1a;


  • Broker&＃xff1a;主机

  • Exchange&＃xff1a;交换机

  • Queue&＃xff1a;队列

  • Connection&＃xff1a;TCP长连接

  • Message&＃xff1a;消息

  • Channel&＃xff1a;消息信道

  • VHost&＃xff1a;虚拟主机

  • Consumer&＃xff1a;消费者

  • Producer&＃xff1a;生产者



Pt1.1 Broker

Broker翻译成中文意思是经纪人、协商人&＃xff0c;而MQ正是在生产者和消费者之间提供消息的转发、代理工作。RabbitMQ需要安装一个独立运行的服务&＃xff0c;服务器节点就称为Broker&＃xff0c;默认端口是5672。



Pt1.2 Exchange

Exchange的作用是接收生产者发送的消息&＃xff0c;并根据Binding规则将消息路由给服务器中的队列。

通常来说&＃xff0c;生产者发送消息给队列&＃xff0c;然后被某个消费者消费。但是如果消费者比较多呢&＃xff0c;比如广播消息的场景&＃xff0c;生产者需要将消息发送给多个队列&＃xff0c;然后再由多个消费者进行消费。

这个生产者看起来压力就很大啊&＃xff0c;需要将同一条消息多次调用发送给每个队列。先不去考虑这么多次消息发送的一致性问题&＃xff08;要发送10个队列&＃xff0c;9个成功&＃xff0c;1个失败的场景下&＃xff0c;是不断重试&＃xff1f;还是要回滚&＃xff1f;这时候消息发送的状态是算成功&＃xff1f;部分成功&＃xff1f;还是失败呢&＃xff1f;想想如果你再发送MQ消息的时候&＃xff0c;还要让你管理这么多复杂的问题&＃xff0c;头大不大&＃xff1f;&＃xff09;&＃xff0c;光是这么多队列的消息发送&＃xff0c;对于生产者客户端的编码人员就是一种痛苦&＃xff0c;要维护这么长的发送清单&＃xff0c;并且每次新增和修改队列&＃xff0c;都要变更代码&＃xff0c;这还只是一条消息的发送&＃xff0c;你说开发人员难不难&＃xff1f;


RabbitMQ已经考虑到这一点。


RabbitMQ提供了Exchange的组件&＃xff0c;负责消息的路由。

也就是说&＃xff0c;一条消息不管需要发送给多少个队列&＃xff0c;多少个消费者&＃xff0c;生产者只需要将消息发送到Exchange上即可&＃xff08;一次发送&＃xff09;&＃xff0c;由Exchange负责消息的转发。Exchange会跟一个或者多个Queue之间建立绑定关系&＃xff0c;它本身不会存储消息&＃xff0c;收到生产者发送的消息后&＃xff0c;会根据绑定关系发送到对应的队列上&＃xff0c;然后由该队列绑定的消费者完成消费。

Exchange和队列是多对多的绑定关系。建立绑定关系后&＃xff0c;生产者发送到Exchange的消息中会携带标识&＃xff0c;当队列的绑定关系跟标识匹配时&＃xff0c;消息就会发送给队列。


可以使用API或者MQ控台声明Exchange。

 ExchangeDeclare(exchange, type, durable, autoDelete, internal, arguments);

  • exchange&＃xff1a;名称&＃xff1b;

  • type&＃xff1a;类型&＃xff0c;一共有3种&＃xff0c;后面会介绍。

  • durable&＃xff1a;元数据是否持久化&＃xff0c;RabbitMQ关闭后&＃xff0c;没有持久化的Exchange将被清除&＃xff08;因为没有持久化是保存在内存中的&＃xff0c;关闭后内存数据丢失&＃xff09;&＃xff1b;

  • autoDelete&＃xff1a;是否自动删除&＃xff0c;如果没有与之绑定的Queue&＃xff0c;直接删除&＃xff1b;

  • internal&＃xff1a;是否内置的&＃xff0c;如果为true&＃xff0c;只能通过Exchange到Exchange&＃xff1b;

  • arguments&＃xff1a;结构化参数


    • Alternate exchange&＃xff1a;备用交换机&＃xff0c;发送消息的时候根据route key并没有把消息路由到队列中去&＃xff0c;这就会将此消息路由到Alternate Exchange属性指定的Exchange上。


ExchangeType决定了Exchange路由消息的行为&＃xff0c;在RabbitMQ中&＃xff0c;ExchangeType有direct、Fanout和Topic三种&＃xff0c;不同类型的Exchange路由的行为是不一样的。


(1) Direct 直连模式

Exchange和队列使用明确地绑定键&＃xff08;binding key&＃xff09;完成绑定&＃xff0c;生产者发送消息时会携带一个路由键&＃xff0c;当消息的路由键和某个队列的绑定键完全匹配时&＃xff0c;这条消息才会从交换机路由到这个队列上。

多个队列可以使用相同的绑定键。

 直连类型的交换机&＃xff0c;适用于一些业务明确的消息&＃xff0c;建立明确的绑定键。

代码示例&＃xff1a;参照[Pt3.2]



(2) Topic 主题模式

主题模式Exchange和队列绑定时&＃xff0c;可以在绑定键中使用通配符&＃xff0c;RabbitMQ支持两个通配符&＃xff1a;


  • # 代表0个或者多个单词

  • * 代表一个单词&＃xff0c;不多不少一个单词

单词&＃xff08;word&＃xff09;指的是使用英文的点“.”隔开的字符。例如&＃xff0c;how.are.you就是3个单词&＃xff0c;im.fine是2个单词。

 主体模式的交换机适用于一些根据业务主题或者消息等级过滤消息的场景&＃xff0c;比如&＃xff0c;一条消息可能跟资金相关、跟风控相关、又跟账户相关&＃xff0c;那就让这个消息指定一个多级的路由键&＃xff0c;第一个单词代表资金&＃xff0c;第二个代表风控&＃xff0c;以此类推等&＃xff0c;让每个消费者的队列都能根据不同的绑定键接收到这个主题的消息。

代码示例&＃xff1a;参考[Pt3.3]



(3) Fanout广播模式

广播模式的Exchange与队列绑定时&＃xff0c;不需要指定绑定键。生产者发送消息到广播Exchange上&＃xff0c;也不需要携带路由键。

消息达到Exchange之后&＃xff0c;所有与Exchange绑定的队列&＃xff0c;都会受到相同消息的副本。

 广播模式的Exchange适用于一些通用的业务消息&＃xff08;通知类型&＃xff09;。比如&＃xff0c;当交易数据发生变化时&＃xff0c;需要通知上层应用&＃xff0c;数据中心和各个数据应用系统&＃xff08;反洗钱、反欺诈、风控、合规做数据归集和风险缝隙&＃xff09;&＃xff0c;就可以使用广播模式的Exchange&＃xff0c;将数据变化广播到各个系统&＃xff0c;由各个系统的消费者来消费。

代码示例&＃xff1a;参照[Pt3.4]



Pt1.3 Queue

FIFO的消息队列&＃xff0c;用于存储还未被消费者消费的消息。

队列是生产者和消费者之间的纽带&＃xff0c;队列的消息被持久化到一个叫做Mnesia的数据库中&＃xff0c;存放在磁盘上。


可以使用API或者在MQ控台声明一个队列&＃xff1a;

 queueDeclare(String queue&＃xff0c; boolean durable&＃xff0c; boolean exclusive&＃xff0c; boolean autoDelete&＃xff0c; Map arguments);

  • queue&＃xff1a; 队列名称&＃xff1b;

  • durable&＃xff1a; 是否持久化&＃xff0c; 队列的声明默认是存放到内存中的&＃xff0c;如果rabbitmq重启会丢失&＃xff0c;如果想重启之后还存在就要使队列持久化&＃xff0c;保存到Erlang自带的Mnesia数据库中&＃xff0c;当rabbitmq重启之后会读取该数据库&＃xff1b;

  • exclusive&＃xff1a;是否排外的&＃xff0c;有两个作用&＃xff0c;一&＃xff1a;当连接关闭时connection.close()该队列是否会自动删除&＃xff1b;二&＃xff1a;该队列是否是私有的private&＃xff0c;如果不是排外的&＃xff0c;可以使用两个消费者都访问同一个队列&＃xff0c;没有任何问题&＃xff0c;如果是排外的&＃xff0c;会对当前队列加锁&＃xff0c;其他通道channel是不能访问的&＃xff0c;如果强制访问会报异常&＃xff1a;com.rabbitmq.client.ShutdownSignalException&＃xff1a; channel error; protocol method&＃xff1a; #method(reply-code&＃61;405&＃xff0c; reply-text&＃61;RESOURCE_LOCKED - cannot obtain exclusive access to locked queue &＃39;queue_name&＃39; in vhost &＃39;/&＃39;&＃xff0c; class-id&＃61;50&＃xff0c; method-id&＃61;20)一般等于true的话用于一个队列只能有一个消费者来消费的场景&＃xff1b;

  • autoDelete&＃xff1a;是否自动删除&＃xff0c;当最后一个消费者断开连接之后队列是否自动被删除&＃xff0c;可以通过RabbitMQ Management&＃xff0c;查看某个队列的消费者数量&＃xff0c;当consumers &＃61; 0时队列就会自动删除&＃xff1b;

  • arguments&＃xff1a;队列中消息处理参数。


    • Message TTL(x-message-ttl)&＃xff1a;设置队列中的所有消息的生存周期(统一为整个队列的所有消息设置生命周期)&＃xff0c; 也可以在发布消息的时候单独为某个消息指定剩余生存时间&＃xff0c;单位毫秒&＃xff0c; 类似于redis中的ttl。生存时间到了&＃xff0c;消息会被从队里中删除&＃xff0c;注意是消息被删除&＃xff0c;而不是队列被删除。 特性Features&＃61;TTL&＃xff0c; 单独为某条消息设置过期时间。

       AMQP.BasicProperties.Builder properties &＃61; new AMQP.BasicProperties().builder().expiration(“6000”);
       channel.basicPublish(EXCHANGE_NAME&＃xff0c; “”&＃xff0c; properties.build()&＃xff0c; message.getBytes(“UTF-8”));

    • Auto Expire(x-expires)&＃xff1a; 当队列在指定的时间没有被访问(consume&＃xff0c; basicGet&＃xff0c; queueDeclare…)就会被删除&＃xff1b;

    • Max Length(x-max-length)&＃xff1a; 限定队列的消息的最大值长度&＃xff0c;超过指定长度将会把最早的几条删除掉&＃xff0c; 类似于mongodb中的固定集合&＃xff0c;例如保存最新的100条消息&＃xff1b;

    • Max Length Bytes(x-max-length-bytes)&＃xff1a; 限定队列最大占用的空间大小&＃xff0c; 一般受限于内存、磁盘的大小&＃xff1b;

    • Overflow Behanviour&＃xff1a;设置队列溢出行为。这决定了当达到队列的最大长度时&＃xff0c;消息会发生什么。有效值为Drop Head或Reject Publish&＃xff1b;

    • Dead letter exchange(x-dead-letter-exchange)&＃xff1a; 当队列消息长度大于最大长度、或者过期的等&＃xff0c;将从队列中删除的消息推送到指定的交换机中去而不是丢弃掉&＃xff1b;

    • Dead letter routing key(x-dead-letter-routing-key)&＃xff1a;将删除的消息推送到指定交换机的指定路由键的队列中去&＃xff1b;

    • Maximum priority(x-max-priority)&＃xff1a;优先级队列&＃xff0c;声明队列时先定义最大优先级值(定义最大值一般不要太大)&＃xff0c;在发布消息的时候指定该消息的优先级&＃xff0c; 优先级更高&＃xff08;数值更大的&＃xff09;的消息先被消费&＃xff1b;

    • Lazy mode(x-queue-mode&＃61;lazy)&＃xff1a; Lazy Queues&＃xff0c;先将消息保存到磁盘上&＃xff0c;不放在内存中&＃xff0c;当消费者开始消费的时候才加载到内存中&＃xff1b;

    • Master locator(x-queue-master-locator)&＃xff1a;将队列设置为主位置模式&＃xff0c;确定在节点集群上声明时队列主位置所依据的规则。


声明了Exchange和Queue之后&＃xff0c;接下来就是绑定二者之间的关系&＃xff0c;同样可以基于API和MQ控台进行操作&＃xff1a;

 channel.QueueBind(queue, exchange, routingKey, arguments);

  • queue&＃xff1a;队列名称

  • exchange&＃xff1a;交换机名称

  • routingKey&＃xff1a;绑定键

  • arguments&＃xff1a;结构化参数。



Pt1.4 Message

由Header和Body组成&＃xff0c;Header是由生产者添加的各种属性的集合&＃xff0c;包括Message是否被持久化、由哪个Message Queue接受、优先级是多少等。Body是真正需要传输的APP数据。



Pt1.5 Connection

无论是生产者还是消费者&＃xff0c;在收发消息前&＃xff0c;都要和Broker建立一个TCP的长连接。



Pt1.6 Channel

Channel是我们与RabbitMQ打交道的最重要的一个接口&＃xff0c;我们大部分的业务操作是在Channel这个接口中完成的&＃xff0c;包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

如果每一次访问RabbitMQ都建立一个Connection&＃xff0c;在消息量大的时候建立TCP Connection的开销将是巨大的&＃xff0c;效率也较低。Channel是在connection内部建立的逻辑连接&＃xff0c;如果应用程序支持多线程&＃xff0c;通常每个thread创建单独的channel进行通讯&＃xff0c;AMQP method包含了channel id帮助客户端和message broker识别channel&＃xff0c;所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销



Pt1.7 VHost

VHost其实是一个虚拟概念&＃xff0c;类似于权限控制组&＃xff0c;一个Virtual Host里面可以有若干个Exchange和Queue&＃xff0c;但是权限控制的最小粒度是Virtual Host。

每一个RabbitMQ服务器都能创建虚拟消息服务器&＃xff0c;我们称之为虚拟主机VHost。每一个vhost本质上是一个mini版的RabbitMQ服务器&＃xff0c;拥有自己的交换机、队列、绑定关系等&＃xff0c;拥有自己的权限机制。vhost之于Rabbit就像虚拟机之于物理机一样。他们通过在各个实例间提供逻辑上分离&＃xff0c;允许为不同的应用程序安全保密的运行数据&＃xff0c;这就是经常被提到的多租户概念。




多租户技术


多租户技术&＃xff08;英语&＃xff1a;multi-tenancy technology&＃xff09;或称多重租赁技术&＃xff0c;是一种软件架构技术&＃xff0c;它是在探讨与实现如何于多用户的环境下共用相同的系统或程序组件&＃xff0c;并且仍可确保各用户间数据的隔离性。


多租户简单来说是指一个单独的实例可以为多个组织服务。


多租户技术为共用的数据中心内&＃xff0c;如何以单一系统架构与服务&＃xff0c;来提供多个客户端相同甚至可定制化的服务&＃xff0c;并且仍然可以保障客户的数据隔离。一个支持多租户技术的系统需要在设计上对它的数据和配置进行虚拟分区&＃xff0c;从而使系统的每个租户或称组织都能够使用一个单独的系统实例&＃xff0c;并且每个租户都可以根据自己的需求对租用的系统实例进行个性化配置。


多租户技术可以实现多个租户之间共享系统实例&＃xff0c;同时又可以实现租户的系统实例的个性化定制。通过使用多租户技术可以保证系统共性的部分被共享&＃xff0c;个性的部分被单独隔离。通过在多个租户之间的资源复用&＃xff0c;运营管理维护资源&＃xff0c;有效节省开发应用的成本。而且&＃xff0c;在租户之间共享应用程序的单个实例&＃xff0c;可以实现当应用程序升级时&＃xff0c;所有租户可以同时升级。同时&＃xff0c;因为多个租户共享一份系统的核心代码&＃xff0c;因此当系统升级时&＃xff0c;只需要升级相同的核心代码即可。



VHost既能将同一个Rabbit的众多客户区分开来&＃xff0c;又可以避免队列和交换器的命名冲突。RabbitMQ提供了开箱即用的默认的虚拟主机“/”&＃xff0c;如果不需要多个vhost可以直接使用这个默认的vhost&＃xff0c;通过使用缺省的guest用户名和guest密码来访问默认的vhost。



Pt1.8 Consumer

Consumer有两种消费模式。


  • Pull模式&＃xff1a;消息存放在MQ服务端&＃xff0c;消费者通过主动获取的方式从MQ获取消息进行消费。这样可以有效的保护消费者&＃xff0c;根据自己的消费能力决定获取消息的频率&＃xff0c;避免消息在Consumer端产生积压造成服务不可用。但是如果消费者消费能力比较差&＃xff0c;发送到MQ的消息按照消费者自身的消费节奏&＃xff0c;需要等待一段时间才能被消费&＃xff0c;会降低消息的实时性。

    RabbitMQ中使用basicGet()拉取数据。

  • Push模式&＃xff1a;只要生产者发送消息到MQ服务器&＃xff0c;就立马推送给消费者&＃xff0c;消息保存在消费者端&＃xff0c;保证消息的实时性。但是如果消费者自身消费能力不够&＃xff0c;可能会造成消息在客户端积压。Spring AMQP采取的就是push模式&＃xff0c;通过事件机制对队列进行监听&＃xff0c;只要有消息达到队列&＃xff0c;就会触发消费消息的方法。

    RabbitMQ中使用basicConsume()推送消息。


RabbitMQ中Pull和Push模式都有实现&＃xff0c;Kafka和RocketMQ只有Pull模式。

由于队列的FIFO特性&＃xff0c;只有确定消息被消费者接收之后&＃xff0c;Broker才会把消息从数据库删除&＃xff0c;继续投递下一条消息。关于MQ如何确认消费者已经收到消息的机制&＃xff0c;在后面可靠性投递部分会详细说明。


一个消费这可以监听多个队列&＃xff0c;一个队列也可以被多个消费者监听。在生产环境中&＃xff0c;一般建议一个消费者只处理一条队列的消息&＃xff0c;如果需要提升队列的消费能力&＃xff0c;可以增加多个消费者&＃xff0c;MQ会以轮训的方式将消息转发到每个消费者上。


通过basicConsume可以从MQ获取消息进行消费。

 channel.basicConsume(queue, true, new DefaultConsumer(channel)
     &＃64;Override
     public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                byte[] body) throws IOException
         String message &＃61; new String(body, "UTF-8");
   
 ;);

  • queue&＃xff1a;要消费的队列名称

  • body&＃xff1a;消息主体



Pt1.9 Producer

Producer就是消息的生产者&＃xff0c;这个实在没什么好介绍的。

通过BasicPublish可以发送消息&＃xff1a;

 channel.BasicPublish(exchange, routingKey, basicProperties, messageBody);

  • exchange&＃xff1a;发送的交换机

  • routingKey&＃xff1a;消息的路由键

  • basicProperties&＃xff1a;设置消息的Header属性&＃xff0c;比如ttl等&＃xff1b;

  • messageBody&＃xff1a;消息的内容&＃xff1b;




Pt2 安装操作

Pt2.1 Windows安装RabbitMQ


(1) Erlang环境说明

RabbitMQ依赖Erlang语言&＃xff0c;所以要先安装Erlang环境。需要注意的是&＃xff0c;RabbitMQ和Erlang有版本对应关系&＃xff0c;否则可能产生奇怪的异常。

具体对应关系可以参照官方链接&＃xff1a;https&＃xff1a;//www.rabbitmq.com/which-erlang.html

我们选择安装最新版RabbitMQ3.8.16&＃xff0c;对应的Erlang环境选择24.0。



(2) 下载安装Erlang

官方链接&＃xff1a;https&＃xff1a;//www.erlang.org/downloads/24.0

因为是Windows环境下&＃xff0c;选择exe文件下载&＃xff0c;然后一直next安装即可。


安装完成后&＃xff0c;配置Erlang环境变量&＃xff1a;

 ERLANG_HOME&＃61;D&＃xff1a;\\Program Files\\erl-24.0
 path中添加&＃xff1a;%ERLANG_HOME%\\bin;

配置完成后&＃xff0c;cmd下输入erl&＃xff0c;查看安装是否正确&＃xff1a;

 C&＃xff1a;\\Users\\lucas>erl
 Eshell V12.0  (abort with ^G)
 1>

(3) 下载安装RabbitMQ

官方下载链接&＃xff1a;https&＃xff1a;//github.com/rabbitmq/rabbitmq-server/releases/tag/v3.8.16

下载rabbitmq-server-3.8.16.exe&＃xff0c;傻瓜式安装。


配置RabbitMQ环境变量&＃xff1a;

 RABBITMQ_SERVER&＃61;D&＃xff1a;\\Program Files\\RabbitMQ Server\\rabbitmq_server-3.8.16
 path中添加&＃xff1a;%RABBITMQ_SERVER%\\sbin

(4) 启用RabbitMQ插件

在cmd中执行命令启用RabbitMQ插件功能&＃xff0c;进入%RABBITMQ_SERVER%\\sbin目录&＃xff0c;执行以下命令&＃xff1a;



rabbitmq-plugins.bat enable rabbitmq_management


 C&＃xff1a;\\Users\\lucas>"G&＃xff1a;\\Software2018\\Developer\\rabbitmq_server-3.8.16\\sbin\\rabbitmq-plugins.bat" enable rabbitmq_management
 Enabling plugins on node rabbit&＃64;lucas-PC&＃xff1a;
 rabbitmq_management
 The following plugins have been configured&＃xff1a;
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_web_dispatch
 Applying plugin configuration to rabbit&＃64;lucas-PC...
 The following plugins have been enabled&＃xff1a;
  rabbitmq_management
  rabbitmq_management_agent
  rabbitmq_web_dispatch
 set 3 plugins.
 Offline change; changes will take effect at broker restart.

(5) 启动RabbitMQ

 # 启动RabbitMQ在cmd下输入&＃xff1a;
 net start RabbitMQ
 ​
 # 关闭RabbitMQ在cmd下输入&＃xff1a;
 net stop RabbitMQ

如果启动MQ报错&＃xff0c;则可能是安装后MQ已经启动了&＃xff0c;可以执行关闭命令后在重新尝试启动。

如果提示[发生系统错误 5。拒绝访问]&＃xff0c;则是需要以管理员身份运行cmd才可以。



(6) RabbitMQ管理控台

RabbitMQ启动后&＃xff0c;访问管理控台http&＃xff1a;//localhost&＃xff1a;15672/&＃xff0c;默认用户名密码guest/guest。

默认配置文件路径&＃xff1a;C&＃xff1a;\\Users\\用户\\AppData\\Roaming\\RabbitMQ\\advanced.config

默认数据库路径&＃xff1a;C&＃xff1a;\\Users\\用户\\AppData\\Roaming\\RabbitMQ\\db\\rabbit&＃64;用户-PC-mnesia



(7) 初始化RabbitMQ

如果要初始化RabbitMQ&＃xff0c;移除全部数据&＃xff0c;可以执行以下命令&＃xff08;谨慎执行&＃xff09;&＃xff1a;

 rabbitmq-service stop
 rabbitmq-service remove
 rabbitmq-service install
 rabbitmq-service start

Pt2.2 CentOS7安装RabbitMQ集群

我之前买过一个云服务器&＃xff0c;我准备在上面搭建一个3节点的集群&＃xff0c;没有云服务器的可以用VM本地搭建一个。



(1) 安装Docker环境

(1) 更新yum源

 sudo yum update

(2) 添加仓库

 sudo yum-config-manager \\
    --add-repo \\
    https&＃xff1a;//mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo

(3) 查看最新版本

如果之前安装了docker&＃xff0c;需要卸载旧版本

 yum list docker-ce --showduplicates | sort -r

(4) 安装Docker CE版本

 yum install docker-ce -y

(2) 安装RabbitMQ

(1) 拉取RabbitMQ镜像&＃xff08;最新版&＃xff09;

 docker pull rabbitmq

(2) 创建docker网络&＃xff08;让容器可以和主机通信&＃xff09;

 docker network create rabbitmqnet

(3) 创建三个容器&＃xff0c;端口分别是 5673 5674 5675 &＃xff0c;管理端口是 15673 15674 15675

 docker run -d --name&＃61;rabbitmq1 -p 5673&＃xff1a;5672 -p 15673&＃xff1a;15672 -e RABBITMQ_NODENAME&＃61;rabbitmq1 -e RABBITMQ_ERLANG_COOKIE&＃61;&＃39;GUPAOEDUFORBETTERYOU&＃39; -h rabbitmq1 --net&＃61;rabbitmqnet rabbitmq&＃xff1a;management
 ​
 docker run -d --name&＃61;rabbitmq2 -p 5674&＃xff1a;5672 -p 15674&＃xff1a;15672 -e RABBITMQ_NODENAME&＃61;rabbitmq1 -e RABBITMQ_ERLANG_COOKIE&＃61;&＃39;GUPAOEDUFORBETTERYOU&＃39; -h rabbitmq2 --net&＃61;rabbitmqnet rabbitmq&＃xff1a;management
 ​
 docker run -d --name&＃61;rabbitmq3 -p 5675&＃xff1a;5672 -p 15675&＃xff1a;15672 -e RABBITMQ_NODENAME&＃61;rabbitmq1 -e RABBITMQ_ERLANG_COOKIE&＃61;&＃39;GUPAOEDUFORBETTERYOU&＃39; -h rabbitmq3 --net&＃61;rabbitmqnet rabbitmq&＃xff1a;management

(4) 后两个节点作为内存节点加入集群

 docker exec -it rabbitmq2 /bin/bash
 rabbitmqctl stop_app
 rabbitmqctl reset
 rabbitmqctl join_cluster --ram rabbitmq1&＃64;rabbitmq1
 rabbitmqctl start_app
 ​
 docker exec -it rabbitmq3 /bin/bash
 rabbitmqctl stop_app
 rabbitmqctl reset
 rabbitmqctl join_cluster --ram rabbitmq1&＃64;rabbitmq1
 rabbitmqctl start_app

结果如下&＃xff1a;

 [root&＃64;VM-0-17-centos ~]# docker exec -it rabbitmq3 /bin/bash
 root&＃64;rabbitmq3&＃xff1a;/# rabbitmqctl stop_app
 RABBITMQ_ERLANG_COOKIE env variable support is deprecated and will be REMOVED in a future version. Use the $HOME/.erlang.COOKIE file or the --erlang-COOKIE switch instead.
 Stopping rabbit application on node rabbitmq1&＃64;rabbitmq3 ...
 root&＃64;rabbitmq3&＃xff1a;/# rabbitmqctl reset
 RABBITMQ_ERLANG_COOKIE env variable support is deprecated and will be REMOVED in a future version. Use the $HOME/.erlang.COOKIE file or the --erlang-COOKIE switch instead.
 Resetting node rabbitmq1&＃64;rabbitmq3 ...
 root&＃64;rabbitmq3&＃xff1a;/# rabbitmqctl join_cluster --ram rabbitmq1&＃64;rabbitmq1
 RABBITMQ_ERLANG_COOKIE env variable support is deprecated and will be REMOVED in a future version. Use the $HOME/.erlang.COOKIE file or the --erlang-COOKIE switch instead.
 Clustering node rabbitmq1&＃64;rabbitmq3 with rabbitmq1&＃64;rabbitmq1
 root&＃64;rabbitmq3&＃xff1a;/# rabbitmqctl start_app
 RABBITMQ_ERLANG_COOKIE env variable support is deprecated and will be REMOVED in a future version. Use the $HOME/.erlang.COOKIE file or the --erlang-COOKIE switch instead.
 Starting node rabbitmq1&＃64;rabbitmq3 ...

(5) 访问控制台

通过http://ip:15673/访问RabbitMQ控台&＃xff0c;默认用户名密码guest/guest




Pt3 基本使用

Pt3.1 基于JAVAAPI示例

(1) 创建Maven项目&＃xff0c;引入RabbitMQ依赖&＃xff1a;

 com.rabbitmq
     amqp-client
     5.12.0
 

(2) 在RabbitMQ控台创建VHost、Exchange、Queue&＃xff0c;以及建立二者绑定关系&＃xff0c;如下图所示&＃xff1a;


  • 添加VHost



  • 添加Exchange



  • 添加Queue



  • 绑定Exchange和Queue


(3) 编写生产者代码

 public class FirstProducer
 ​
     private static final String FIRST_MSG &＃61; "This is my first rabbitmq message.";
 ​
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 发送消息
             // RabbitMQ的API都是基于channel的
             channel.basicPublish(RabbitMQConstants.EXCH_FIRST_EXCHANGE&＃xff0c; "first"&＃xff0c; null&＃xff0c; FIRST_MSG.getBytes());
             System.out.println("消息发送成功。");
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 

运行生产者&＃xff0c;可以看到队列中已经收到了消息&＃xff0c;但是我们此时还没有建立消费者&＃xff0c;所以消息保存在内存中没有被消费。


(4) 编写消费者代码

 public class FirstConsumer
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 前面已经在控台绑定了Exchange和Queue&＃xff0c;这里不需要重复操作。如果是不操作控台&＃xff0c;只是用API&＃xff0c;可以使用下列API创建Queue并绑定Exchange
             // 声明Exchange
             // channel.exchangeDeclare(RabbitMQConstants.EXCH_FIRST_EXCHANGE&＃xff0c; "direct"&＃xff0c; false&＃xff0c; false&＃xff0c; null);
             // 声明队列
             // channel.queueDeclare(RabbitMQConstants.QUEUE_FIRST_QUEUE&＃xff0c; false&＃xff0c; false&＃xff0c; false&＃xff0c; null);
             // 绑定队列和交换机
             // channel.queueBind(RabbitMQConstants.QUEUE_FIRST_QUEUE&＃xff0c; RabbitMQConstants.EXCH_FIRST_EXCHANGE&＃xff0c; "first");
 ​
             // 创建消费者&＃xff0c;绑定队列
             Consumer consumer &＃61; new DefaultConsumer(channel)
                 &＃64;Override
                 public void handleDelivery(String consumerTag&＃xff0c; Envelope envelope&＃xff0c; AMQP.BasicProperties properties&＃xff0c;
                     byte[] body) throws IOException
                     String message &＃61; new String(body&＃xff0c; "UTF-8");
                     System.out.println("收到消息&＃xff1a;" &＃43; message);
               
            ;
 ​
             while (true)
                 // 消费者接受Queue消息
                 channel.basicConsume(RabbitMQConstants.QUEUE_FIRST_QUEUE&＃xff0c; true&＃xff0c; consumer);
           
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 

测试结果&＃xff1a;

 收到消息&＃xff1a;This is my first rabbitmq message.
 收到消息&＃xff1a;This is my first rabbitmq message.

Pt3.2 Direct模式

在示例中&＃xff0c;Exchange、Queue和绑定关系都是通过API来完成&＃xff0c;在实际生产应用中&＃xff0c;建议使用控台&＃xff0c;这样可以有效进行权限控制&＃xff0c;并且可以增加审核机制&＃xff0c;避免硬代码。

 // 创建生产者
 public class DirectProducer
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明Exchange
             channel.exchangeDeclare("DIRECT_EXCHANGE"&＃xff0c; "direct"&＃xff0c; false&＃xff0c; false&＃xff0c; null);
 ​
             // 发送消息&＃xff0c;路由键不同
             channel.basicPublish("DIRECT_EXCHANGE"&＃xff0c; "first"&＃xff0c; null&＃xff0c; "first message".getBytes());
             channel.basicPublish("DIRECT_EXCHANGE"&＃xff0c; "second"&＃xff0c; null&＃xff0c; "second message".getBytes());
             channel.basicPublish("DIRECT_EXCHANGE"&＃xff0c; "third"&＃xff0c; null&＃xff0c; "third message".getBytes());
             System.out.println("消息发送成功。");
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 

 // 创建消费者&＃xff1a;一共创建3个消费者&＃xff0c;绑定同一个Exchange&＃xff0c;绑定键不同。
 public class DirectConsumer1
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明队列
             channel.queueDeclare("DIRECT_FIRST_QUEUE"&＃xff0c; false&＃xff0c; false&＃xff0c; false&＃xff0c; null);
             // 绑定队列和交换机
             channel.queueBind("DIRECT_FIRST_QUEUE"&＃xff0c; "DIRECT_EXCHANGE"&＃xff0c; "first");
 ​
             // 创建消费者&＃xff0c;绑定队列
             Consumer consumer &＃61; new DefaultConsumer(channel)
                 &＃64;Override
                 public void handleDelivery(String consumerTag&＃xff0c; Envelope envelope&＃xff0c; AMQP.BasicProperties properties&＃xff0c;
                     byte[] body) throws IOException
                     String message &＃61; new String(body&＃xff0c; "UTF-8");
                     System.out.println("消费者1收到消息&＃xff1a;" &＃43; message);
               
            ;
 ​
             while (true)
                 // 消费者接受Queue消息
                 channel.basicConsume("DIRECT_FIRST_QUEUE"&＃xff0c; true&＃xff0c; consumer);
           
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 
 ​
 ​
 public class DirectConsumer2
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明队列
             channel.queueDeclare("DIRECT_SECOND_QUEUE"&＃xff0c; false&＃xff0c; false&＃xff0c; false&＃xff0c; null);
             // 绑定队列和交换机
             channel.queueBind("DIRECT_SECOND_QUEUE"&＃xff0c; "DIRECT_EXCHANGE"&＃xff0c; "second");
 ​
             // 创建消费者&＃xff0c;绑定队列
             Consumer consumer &＃61; new DefaultConsumer(channel)
                 &＃64;Override
                 public void handleDelivery(String consumerTag&＃xff0c; Envelope envelope&＃xff0c; AMQP.BasicProperties properties&＃xff0c;
                     byte[] body) throws IOException
                     String message &＃61; new String(body&＃xff0c; "UTF-8");
                     System.out.println("消费者2收到消息&＃xff1a;" &＃43; message);
               
            ;
 ​
             while (true)
                 // 消费者接受Queue消息
                 channel.basicConsume("DIRECT_SECOND_QUEUE"&＃xff0c; true&＃xff0c; consumer);
           
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 
 ​
 ​
 public class DirectConsumer3
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明队列
             channel.queueDeclare("DIRECT_THIRD_QUEUE"&＃xff0c; false&＃xff0c; false&＃xff0c; false&＃xff0c; null);
             // 绑定队列和交换机
             channel.queueBind("DIRECT_THIRD_QUEUE"&＃xff0c; "DIRECT_EXCHANGE"&＃xff0c; "third");
 ​
             // 创建消费者&＃xff0c;绑定队列
             Consumer consumer &＃61; new DefaultConsumer(channel)
                 &＃64;Override
                 public void handleDelivery(String consumerTag&＃xff0c; Envelope envelope&＃xff0c; AMQP.BasicProperties properties&＃xff0c;
                     byte[] body) throws IOException
                     String message &＃61; new String(body&＃xff0c; "UTF-8");
                     System.out.println("消费者3收到消息&＃xff1a;" &＃43; message);
               
            ;
 ​
             while (true)
                 // 消费者接受Queue消息
                 channel.basicConsume("DIRECT_THIRD_QUEUE"&＃xff0c; true&＃xff0c; consumer);
           
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 

分别运行生产者和消费者&＃xff1a;

 // 生产和
 消息发送成功。
 ​
 // 消费者1
 消费者1收到消息&＃xff1a;first message
 ​
 // 消费者2
 消费者2收到消息&＃xff1a;second message
 ​
 // 消费者3
 消费者3收到消息&＃xff1a;third message

Pt3.3 Topic模式

在示例中&＃xff0c;Exchange、Queue和绑定关系都是通过API来完成&＃xff0c;在实际生产应用中&＃xff0c;建议使用控台&＃xff0c;这样可以有效进行权限控制&＃xff0c;并且可以增加审核机制&＃xff0c;避免硬代码。

 // 创建生产者
 public class TopicProducer
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明Exchange
             channel.exchangeDeclare("TOPIC_EXCHANGE"&＃xff0c; "topic"&＃xff0c; false&＃xff0c; false&＃xff0c; null);
 ​
             // 发送消息&＃xff1a;Topic类型不用路由键
             channel.basicPublish("TOPIC_EXCHANGE"&＃xff0c; "money.risk"&＃xff0c; null&＃xff0c; "money.risk".getBytes());
             channel.basicPublish("TOPIC_EXCHANGE"&＃xff0c; "money.risk.account"&＃xff0c; null&＃xff0c; "money.risk.account".getBytes());
             channel.basicPublish("TOPIC_EXCHANGE"&＃xff0c; "risk.account"&＃xff0c; null&＃xff0c; "risk.account".getBytes());
             channel.basicPublish("TOPIC_EXCHANGE"&＃xff0c; "risk.account.order"&＃xff0c; null&＃xff0c; "risk.account".getBytes());
 ​
             System.out.println("消息发送成功。");
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 
 // 创建消费者
 public class TopicConsumer1
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明队列
             channel.queueDeclare("TOPIC_FIRST_QUEUE"&＃xff0c; false&＃xff0c; false&＃xff0c; false&＃xff0c; null);
             // 绑定队列和交换机
             channel.queueBind("TOPIC_FIRST_QUEUE"&＃xff0c; "TOPIC_EXCHANGE"&＃xff0c; "money.#");
 ​
             // 创建消费者&＃xff0c;绑定队列
             Consumer consumer &＃61; new DefaultConsumer(channel)
                 &＃64;Override
                 public void handleDelivery(String consumerTag&＃xff0c; Envelope envelope&＃xff0c; AMQP.BasicProperties properties&＃xff0c;
                     byte[] body) throws IOException
                     String message &＃61; new String(body&＃xff0c; "UTF-8");
                     System.out.println("消费者1收到消息&＃xff1a;" &＃43; message);
               
            ;
 ​
             while (true)
                 // 消费者接受Queue消息
                 channel.basicConsume("TOPIC_FIRST_QUEUE"&＃xff0c; true&＃xff0c; consumer);
           
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 
 ​
 ​
 public class TopicConsumer2
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明队列
             channel.queueDeclare("TOPIC_SECOND_QUEUE"&＃xff0c; false&＃xff0c; false&＃xff0c; false&＃xff0c; null);
             // 绑定队列和交换机
             channel.queueBind("TOPIC_SECOND_QUEUE"&＃xff0c; "TOPIC_EXCHANGE"&＃xff0c; "*.risk.#");
 ​
             // 创建消费者&＃xff0c;绑定队列
             Consumer consumer &＃61; new DefaultConsumer(channel)
                 &＃64;Override
                 public void handleDelivery(String consumerTag&＃xff0c; Envelope envelope&＃xff0c; AMQP.BasicProperties properties&＃xff0c;
                     byte[] body) throws IOException
                     String message &＃61; new String(body&＃xff0c; "UTF-8");
                     System.out.println("消费者2收到消息&＃xff1a;" &＃43; message);
               
            ;
 ​
             while (true)
                 // 消费者接受Queue消息
                 channel.basicConsume("TOPIC_SECOND_QUEUE"&＃xff0c; true&＃xff0c; consumer);
           
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 
 ​
 ​
 public class TopicConsumer3
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明队列
             channel.queueDeclare("TOPIC_THIRD_QUEUE"&＃xff0c; false&＃xff0c; false&＃xff0c; false&＃xff0c; null);
             // 绑定队列和交换机
             channel.queueBind("TOPIC_THIRD_QUEUE"&＃xff0c; "TOPIC_EXCHANGE"&＃xff0c; "#.account");
 ​
             // 创建消费者&＃xff0c;绑定队列
             Consumer consumer &＃61; new DefaultConsumer(channel)
                 &＃64;Override
                 public void handleDelivery(String consumerTag&＃xff0c; Envelope envelope&＃xff0c; AMQP.BasicProperties properties&＃xff0c;
                     byte[] body) throws IOException
                     String message &＃61; new String(body&＃xff0c; "UTF-8");
                     System.out.println("消费者3收到消息&＃xff1a;" &＃43; message);
               
            ;
 ​
             while (true)
                 // 消费者接受Queue消息
                 channel.basicConsume("TOPIC_THIRD_QUEUE"&＃xff0c; true&＃xff0c; consumer);
           
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 

运行结果&＃xff1a;

 // 生产者输出
 消息发送成功。
 ​
 // 消费者1输出
 消费者1收到消息&＃xff1a;money.risk
 消费者1收到消息&＃xff1a;money.risk.account
 ​
 // 消费者2输出
 消费者2收到消息&＃xff1a;money.risk
 消费者2收到消息&＃xff1a;money.risk.account
 ​
 // 消费者3输出
 消费者3收到消息&＃xff1a;money.risk.account
 消费者3收到消息&＃xff1a;risk.account

Pt3.4 Fanout模式

在示例中&＃xff0c;Exchange、Queue和绑定关系都是通过API来完成&＃xff0c;在实际生产应用中&＃xff0c;建议使用控台&＃xff0c;这样可以有效进行权限控制&＃xff0c;并且可以增加审核机制&＃xff0c;避免硬代码。

 // 创建生产者
 public class FanoutProducer
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配置连接工厂类参数
             ConnectionFactory factory &＃61; new ConnectionFactory();
             factory.setHost(RabbitMQConstants.HOST);
             factory.setPort(RabbitMQConstants.PORT);
             factory.setVirtualHost(RabbitMQConstants.VHOST);
             factory.setUsername(RabbitMQConstants.USERNAME);
             factory.setPassword(RabbitMQConstants.PASSWORD);
 ​
             // 与RabbitMQ服务器建立连接
             connection &＃61; factory.newConnection();
             channel &＃61; connection.createChannel();
 ​
             // 声明Exchange
             channel.exchangeDeclare("FANOUT_EXCHANGE"&＃xff0c; "fanout"&＃xff0c; false&＃xff0c; false&＃xff0c; null);
 ​
             // 发送消息&＃xff1a;FANOUT类型不用路由键
             channel.basicPublish("FANOUT_EXCHANGE"&＃xff0c; ""&＃xff0c; null&＃xff0c; "first message".getBytes());
             channel.basicPublish("FANOUT_EXCHANGE"&＃xff0c; ""&＃xff0c; null&＃xff0c; "second message".getBytes());
             channel.basicPublish("FANOUT_EXCHANGE"&＃xff0c; ""&＃xff0c; null&＃xff0c; "third message".getBytes());
             System.out.println("消息发送成功。");
        catch (Exception ex)
             ex.printStackTrace();
        finally
             try
                 channel.close();
                 connection.close();
            catch (Exception ex)
                 ex.printStackTrace();
           
       
   
 
 // 创建消费者
 public class FanoutConsumer1
     public static void main(String[] args)
 ​
         Connection connection &＃61; null;
         Channel channel &＃61; null;
 ​
         try
             // 配

推荐阅读
author-avatar
冠臻祥勇8522
这个家伙很懒，什么也没留下！
Tags | 热门标签
RankList | 热门文章
PHP1.CN | 中国最专业的PHP中文社区 | DevBox开发工具箱 | json解析格式化 |PHP资讯 | PHP教程 | 数据库技术 | 服务器技术 | 前端开发技术 | PHP框架 | 开发工具 | 在线工具
Copyright © 1998 - 2020 PHP1.CN. All Rights Reserved | 京公网安备 11010802041100号 | 京ICP备19059560号-4 | PHP1.CN 第一PHP社区 版权所有